package com.vodafone.mapreduce.map;

import java.util.List;
import java.util.Map;

public class MapProcessor<K, V, L, W> extends Thread {

    private MapFunction<K, V, L, W> mapFunction;
    private CombineFunction<K, V, L, W> combineFunction;

    public void setMapFunction(MapFunction<K, V, L, W> mapFunction) {
        this.mapFunction = mapFunction;
    }

    public void setCombineFunction(CombineFunction<K, V, L, W> combineFunction) {
        this.combineFunction = combineFunction;
    }

    public Map<L, List<W>> processMapMessage(K k, V v) {
        mapFunction.initialize(k, v);
        mapFunction.map();
        combineFunction.initialize(mapFunction.getMapResult());
        combineFunction.combine();
        return combineFunction.getCombineResult();
    }

}
